package com.bigdata.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 窗口函数
 */
object Demo7Window {

  def main(args: Array[String]): Unit = {
    /**
     * 窗口函数
     * row_number
     * rank
     * dense_rank
     * sum
     * count
     * avg
     * max
     * min
     * lag 取当前行前面
     * lead： 取当前行后面的
     * first_value
     * last_value
     */
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("window")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    spark.sparkContext.setLogLevel("error")

    //学生表
    val studentDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
      .load("data/students.txt")

    //分数表 sid 学生id， cid 课程id，sco 得分
    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("sid STRING,cid STRING,sco DOUBLE")
      .load("data/score.txt")

    //科目表 ssco表示科目的满分
    val subjectDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("cid STRING,cname STRING,ssco DOUBLE")
      .load("data/subject.txt")

    // 把学生信息关联分数表，得到学生分数表信息，默认是inner join
    val joinDF: DataFrame = studentDF.join(scoreDF, $"id" === $"sid")

    /**
     * 1、统计  统计每个班级的学生总分排名，得到排名前十的学生
     * 年级 整个文科、理科的总分排名
     * sum over 由两种用法
     * 1、分区不排序----总和
     * 2、分区也排序-----累计求和
     */
      joinDF
        .groupBy($"clazz", $"id", $"name")
        .agg(sum($"sco") as "sum_sco")
        .withColumn("r", row_number() over Window.partitionBy($"clazz").orderBy($"sum_sco".desc))
        .where($"r" <= 10)
        // .show(30)

    //统计  所有学生总分，按照总分倒序排列，取前十名
    joinDF
      // 按照学生id分区，统计的是当前学生的所有学科总分
      .withColumn("sumSco", sum($"sco") over Window.partitionBy($"id"))
      .groupBy($"id", $"name")
      .agg(max($"sumSco") as "total_sco")
      // 按照总分倒序，总分一致的话，按照id倒序
      .orderBy($"total_sco".desc)
      .limit(10)
     // .show()

    /**
     * 3、统计每科都及格的学生
     */
    val fullDF: DataFrame = joinDF.join(subjectDF, "cid")

    fullDF
      .withColumn("flag", when($"sco" >= $"ssco" * 0.6, 0).otherwise(1))
      .withColumn("sum_flag", sum($"flag") over Window.partitionBy($"id"))
      .where($"sum_flag" === 0)
      .groupBy($"id",$"name")
      .agg($"id" as  "new_id")
     // .show()

    /**
     * 2、统计总分    大于班級平均分的    学生每个科目的成绩
     */
    joinDF
      // 统计每个学生的总分
      .withColumn("sum_sco", sum($"sco") over Window.partitionBy($"id"))
      .withColumn("avg_sco", avg($"sco") over Window.partitionBy($"clazz"))
      .where($"sum_sco" > $"avg_sco")
       // .show(100,false)


    /**
     * 统计每个班级每个学生名次分数差
     * lag:取当前行前面的数据，需要分区和排序
     * lead 取当前行后面的数据，需要分区和排序
     */
    joinDF
      .groupBy($"clazz",$"name", $"id")
      .agg(sum($"sco") as "sumSco")
      .withColumn("headSumSco",
        lag($"sumSco", 1, 750) over Window.partitionBy($"clazz").orderBy($"sumSco".desc))
      .withColumn("cha", $"headSumSco" - $"sumSco")
      .show()


  }

}
